package org.swift.mashup.engine.concurrency;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.log4j.Logger;
import org.swift.mashup.engine.MashupException;
import org.swift.mashup.engine.concurrency.agenda.Agenda;
import org.swift.mashup.engine.concurrency.agenda.AgendaItem;
import org.swift.mashup.engine.concurrency.agenda.ConcurrentAgenda;

public abstract class ConcurrentContext {
	private static final Logger log = Logger.getLogger(ConcurrentContext.class);
	
	private String sessionID;
	private ConcurrentConfiguration configuration;

	private List<AgendaItem> jobs = new ArrayList<AgendaItem>();
	private int totalJobs = 0;
	private int processedJobs = 0;

	// keeps track to the last compute node that was used.
	private Iterator<String> computeNodeIterator;

	public ConcurrentContext(String sessionId,ConcurrentConfiguration configuration) {
		this.sessionID = sessionId;
		this.configuration = configuration;
		System.setProperty("org.swift.mashup.engine.concurrency.WorkingDirectory",
			configuration.getWorkingDirectory());
	}

	public ConcurrentContext(String sessionid, String workingDirectory) throws MashupException {
		this.sessionID = sessionid;
		this.configuration = new ConcurrencySettingsLoader(workingDirectory,true);
		System.setProperty("org.swift.mashup.engine.concurrency.WorkingDirectory",
			configuration.getWorkingDirectory());
	}

	public String getWorkingDirectory() {
		return configuration.getWorkingDirectory();
	}

	public String getSessionID() {
		return sessionID;
	}

	public int getTotalJobs() {
		return this.totalJobs;
	}

	public void runJobs() {
		Agenda agenda = new ConcurrentAgenda(this.jobs);
		agenda.run();
		this.jobs.clear();
	}

	/**
	 * Creates a new jobset using a givin processid
	 * 
	 * @param processid
	 * @param startIndex
	 * @param endIndex
	 */
	public void createJobSet(String processid, int startIndex, int interval) throws MashupException {
		log.debug("Creating job for process: :" + processid + "start: "
			+ startIndex + "interval:" + interval);
		this.jobs.add(new AgendaItem(this.createJob(processid, getWorkingDirectory(), 
			startIndex,	interval), "Job[" + this.totalJobs + "]"));
	}

	/**
	 * Creates a new jobset with a new process id
	 * 
	 * @param startIndex
	 * @param endIndex
	 */
	public void createJobSet(int startIndex, int interval) throws MashupException {
		final String processid = String.valueOf(++this.totalJobs);
		this.createJobSet(processid, startIndex, interval);
	}

	public synchronized String getNextComputeNode() throws ConcurrencyException {
		if ((computeNodeIterator == null || !computeNodeIterator.hasNext()) 
				&& getComputeNodeCount() > 0){
			// reset the iterator
			computeNodeIterator = configuration.getComputeNodeIterator();
		} 
		if (!computeNodeIterator.hasNext()){
			throw new ConcurrencyException ("No compute nodes are available.");
		}
		return computeNodeIterator.next();
	}

	/**
	 * Called to create jobs to put into the jobQueue
	 * 
	 * @param processid
	 * @param sessionid
	 * @param workingDirectory
	 * @param startIndex
	 * @param interval
	 * @return
	 */
	protected abstract Runnable createJob(String processid, String workingDirectory,
			int startIndex, int interval);

	// Helpers for creating your own load balancer
	public int getJobQueueSize() {
		return this.jobs.size();
	}

	public int getComputeNodeCount() {
		return configuration.computeNodeCount();
	}

	public int getProcessedJobs() {
		return processedJobs;
	}

	public void setProcessedJobs(int processedJobs) {
		this.processedJobs = processedJobs;
	}
	
	public ConcurrentConfiguration getConfiguration (){
		return this.configuration;
	}
	public boolean isDeleteFiles(){
		return true;
	}
	public boolean isMappedFiles(){
		return true;
	}

}
